package com.demo.mq.rocket.provider.schedule;

import com.alibaba.fastjson.JSON;
import javafx.util.Pair;
import lombok.Data;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import static com.demo.mq.rocket.provider.Const.ROCKET_DEFAULT_GROUP;
import static com.demo.mq.rocket.provider.Const.ROCKET_NAMESRVADDR;

/**
 * 开源版本的RocketMQ中，对延迟消息并不支持任意时间的延迟设定(商业版本中支持)，
 * 而是只支持18个固定的延迟级别，
 * 1到18分别对应messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
 * <p>
 * 这里实现支持任意时间
 */
public class DelayMessageProducer {
    private static final String schedule_topic = "self_schedule_topic";

    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer(ROCKET_DEFAULT_GROUP);
        producer.setNamesrvAddr(ROCKET_NAMESRVADDR);
        producer.start();

        SelfScheduleConsumer consumer = new SelfScheduleConsumer(producer);

        String msg = "我是延迟消息:" + LocalDateTime.now();
        DelayMessage delayMessage = new DelayMessage(msg, "simple_topic", 13, TimeUnit.SECONDS);

        sendDelayMessage(producer, delayMessage);


        TimeUnit.HOURS.sleep(1);
    }

    private static void sendDelayMessage(DefaultMQProducer producer, DelayMessage delayMessage) throws Exception {
        int delayTimeLevel = delayTimeLevel(delayMessage.getActionTime());
        Message message;
        if (delayTimeLevel > 0) {
            message = new Message(schedule_topic, delayMessage.toString().getBytes(StandardCharsets.UTF_8));
            // 唯一区别:设置消息延迟级别
            message.setDelayTimeLevel(delayTimeLevel);
            producer.send(message);
        } else {
            message = new Message(delayMessage.getTopic(), delayMessage.getMsg().getBytes(StandardCharsets.UTF_8));
            producer.send(message);
        }

    }

    // 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
    private static final List<Pair<Integer, Integer>> DELAY_TIME_LEVEL_DATA = new ArrayList<Pair<Integer, Integer>>() {{
        add(new Pair<>(18, 2 * 60 * 60));
        add(new Pair<>(17, 60 * 60));
        add(new Pair<>(16, 30 * 60));
        add(new Pair<>(15, 20 * 60));
        add(new Pair<>(14, 10 * 60));
        add(new Pair<>(13, 9 * 60));
        add(new Pair<>(12, 8 * 60));
        add(new Pair<>(11, 7 * 60));
        add(new Pair<>(10, 6 * 60));
        add(new Pair<>(9, 5 * 60));
        add(new Pair<>(8, 4 * 60));
        add(new Pair<>(7, 3 * 60));
        add(new Pair<>(6, 2 * 60));
        add(new Pair<>(5, 60));
        add(new Pair<>(4, 30));
        add(new Pair<>(3, 10));
        add(new Pair<>(2, 5));
        add(new Pair<>(1, 1));
    }};


    private static int delayTimeLevel(LocalDateTime actionTime) {
        long seconds = ChronoUnit.SECONDS.between(LocalDateTime.now(), actionTime);
        if (seconds <= 0) {
            return 0;
        }
        for (Pair<Integer, Integer> pair : DELAY_TIME_LEVEL_DATA) {
            if (seconds >= pair.getValue()) {
                return pair.getKey();
            }
        }
        return 0;
    }

    /**
     * 实现任意时间的延迟设定
     */
    static class SelfScheduleConsumer {

        private final DefaultMQPushConsumer consumer;
        private final DefaultMQProducer producer;

        public SelfScheduleConsumer(DefaultMQProducer producer) throws Exception {
            this.producer = producer;
            consumer = new DefaultMQPushConsumer(ROCKET_DEFAULT_GROUP);
            consumer.setNamesrvAddr(ROCKET_NAMESRVADDR);
            consumer.subscribe(schedule_topic, "*");
            run();
            System.out.printf("消费者已启动.%n");
        }

        private void run() throws MQClientException {
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                                ConsumeConcurrentlyContext context) {
                    try {
                        for (MessageExt msg : msgs) {
                            DelayMessage delayMessage = JSON.toJavaObject(JSON.parseObject(new String(msg.getBody())), DelayMessage.class);
                            System.out.println(delayMessage);
                            sendDelayMessage(producer, delayMessage);
                        }
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    } catch (Exception e) {
                        e.printStackTrace();
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
            });

            consumer.start();
        }
    }

    @Data
    static class DelayMessage {
        public DelayMessage() {
        }

        public DelayMessage(String msg, String topic, LocalDateTime actionTime) {
            this.msg = msg;
            this.topic = topic;
            this.actionTime = actionTime;
        }

        public DelayMessage(String msg, String topic, int time, TimeUnit timeUnit) {
            this.msg = msg;
            this.topic = topic;
            this.actionTime = LocalDateTime.now().plusSeconds(TimeUnit.SECONDS.convert(time, timeUnit));
        }

        private String msg;
        private String topic;
        private LocalDateTime actionTime;

        @Override
        public String toString() {
            return JSON.toJSONStringWithDateFormat(this, "yyyy-MM-dd HH:mm:ss");
        }
    }


}
